package org.study.sync;

import com.alibaba.fastjson2.JSON;
import com.google.common.base.Stopwatch;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;
import java.util.function.*;

/**
 * 类的描述
 *
 * @author Administrator
 * @date 2021-08-27
 */
public class CompletableFutureTests1 {
    public static void main(String[] args) {
        try {
            //test01();
            //test02();
            //test03();
            //test04();
            test05_1();
            //test05_2();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

    /**
     * 使用CompletableFuture执行多个异步任务,将结果合并返回
     * https：//blog.csdn.net/zxk1995/article/details/103661187
     *
     * @return java.util.Map<java.lang.String, java.lang.String>
     */
    public static void test01() throws InterruptedException, ExecutionException {
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss:SSS");
        System.out.printf("准备执行任务，当前时间：%s\n", formatter.format(LocalDateTime.now()));
        // 不存在并发插入情况，不需要使用ConcurrentHashMap
        // Map<String, String> data = new ConcurrentHashMap<>(3);
        Map<String, String> data = new HashMap<>(3);
        // 第一个任务
        CompletableFuture<String> task01 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "task01";
        });
        // 第二个任务
        CompletableFuture<String> task02 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "task02";
        });
        // 第三个任务
        CompletableFuture<String> task03 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "task03";
        });
        // get()方法会阻塞
        data.put("task01", task01.get());
        System.out.printf("task01执行完毕，当前时间：%s\n", formatter.format(LocalDateTime.now()));
        data.put("task02", task02.get());
        System.out.printf("task02执行完毕，当前时间：%s\n", formatter.format(LocalDateTime.now()));
        data.put("task03", task03.get());
        System.out.printf("task03执行完毕，当前时间：%s\n", formatter.format(LocalDateTime.now()));
        System.out.println(JSON.toJSONString(data));
    }

    /**
     * Java CompletableFuture：allOf等待所有异步线程任务结束
     * https：//blog.csdn.net/zhangphil/article/details/80670593
     */
    public static void test02() throws ExecutionException, InterruptedException {
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss:SSS");
        System.out.printf("任务开始，当前时间：%s\n", formatter.format(LocalDateTime.now()));
        CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "f1";
        });
        f1.whenCompleteAsync(new BiConsumer<String, Throwable>() {
            @Override
            public void accept(String msg, Throwable throwable) {
                System.out.printf("任务：%s，当前时间：%s\n", msg, formatter.format(LocalDateTime.now()));
            }
        });
        CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "f2";
        });
        f2.whenCompleteAsync(new BiConsumer<String, Throwable>() {
            @Override
            public void accept(String msg, Throwable throwable) {
                System.out.printf("任务：%s，当前时间：%s\n", msg, formatter.format(LocalDateTime.now()));
            }
        });
        CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2);
        // 阻塞，直到所有任务结束
        System.out.printf("阻塞开始，当前时间：%s\n", formatter.format(LocalDateTime.now()));
        all.join();
        System.out.printf("阻塞结束，当前时间：%s\n", formatter.format(LocalDateTime.now()));
        // 一个需要耗时2秒，一个需要耗时3秒，只有当最长的耗时3秒的完成后，才会结束
        System.out.printf("任务结束，当前时间：%s\n", formatter.format(LocalDateTime.now()));
    }

    /**
     * CompletableFuture 合并多个微服务的查询结果
     * https://blog.csdn.net/weixin_33995481/article/details/92635231
     */
    public static void test03() throws ExecutionException, InterruptedException {
        final Stopwatch stopwatch = Stopwatch.createStarted();
        CompletableFuture<Integer> result = CompletableFuture.supplyAsync(() -> {
            try {
                // 模拟微服务A的查询接口
                System.out.println("微服务A的查询接口：" + Thread.currentThread().getName());
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 1;
        }).thenCombineAsync(CompletableFuture.supplyAsync(() -> {
            try {
                // 模拟微服务B的查询接口
                System.out.println("微服务B的查询接口：" + Thread.currentThread().getName());
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 2;
        }), (r1, r2) -> {
            // 合并两个查询结果
            return r1 + r2;
        }).thenCombineAsync(CompletableFuture.supplyAsync(() -> {
            try {
                // 模拟微服务C的查询接口
                System.out.println("微服务C的查询接口：" + Thread.currentThread().getName());
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 3;
        }), (r2, r3) -> {
            // 合并两个查询结果
            return r2 + r3;
        });
        System.out.println("线程：" + Thread.currentThread().getName());
        System.out.println("最终结果：" + result.get());
        stopwatch.stop();
        // 整体请求时间以最长的来定
        System.out.println("总耗时：" + stopwatch.elapsed(TimeUnit.SECONDS));
    }

    /**
     * CompletableFuture thenApplyAsync链式编程
     * https://zhangphil.blog.csdn.net/article/details/90721512
     */
    public static void test04() throws ExecutionException, InterruptedException {
        CompletableFuture future = CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "任务A";
            }
        }).thenApplyAsync(new Function<String, Object>() {
            @Override
            public Object apply(String s) {
                System.out.println(s);
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return "任务B";
            }
        });
        future.thenApplyAsync(new Function<String, Object>() {
            @Override
            public Object apply(String s) {
                System.out.println(s);
                System.out.println("任务C");
                return "任务C";
            }
        });
        System.out.println(future.get());
    }

    /**
     * CompletableFuture 组合处理 allOf 和 anyOf太赞了！
     * https://baijiahao.baidu.com/s?id=1701173396839580698&wfr=spider&for=pc
     * https://www.freesion.com/article/76011072280/
     */
    public static void test05_1() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("future1 finished!");
            return "future1 finished!";
        });
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("future2 finished!");
            return "future2 finished!";
        });
        CompletableFuture<Void> allOf = CompletableFuture.allOf(future1, future2);
        allOf.get();
        System.out.println("future1: " + future1.isDone() + " future2: " + future2.isDone());
    }

    public static void test05_2() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "future1结果";
        });
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "future2结果";
        });
        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "future3结果";
        });
        CompletableFuture<Object> completableFuture = CompletableFuture.anyOf(future1, future2, future3);
        System.out.println(completableFuture.get());
    }
}
